package com.cryptoneedle.center.source.service;

import com.cryptoneedle.center.config.helper.ConfigPropertyHelper;
import com.cryptoneedle.center.source.constant.Constant;
import com.cryptoneedle.center.source.entity.metadata.Catalog;
import com.cryptoneedle.center.source.entity.metadata.Column;
import com.cryptoneedle.center.source.entity.metadata.Table;
import com.cryptoneedle.center.source.enums.DorisDataType;
import com.cryptoneedle.center.source.properties.DorisProperties;
import com.cryptoneedle.center.source.service.crud.*;
import com.cryptoneedle.center.source.util.CryptoUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * <p>description:  </p>
 *
 * @author CryptoNeedle
 * @date 2025-10-23
 */
@Slf4j
@Service
@Transactional
public class ScriptService {

    @Resource
    private ScriptService scriptService;

    @Autowired
    private FoundDataService select;
    @Autowired
    private AddDataService add;
    @Autowired
    private PatchDataService patch;
    @Autowired
    private SaveDataService save;
    @Autowired
    private DeleteDataService delete;

    @Resource
    private DorisProperties dorisProperties;

    public String dorisScript(String catalogName, String databaseName, String tableName) {
        Catalog catalog = select.catalog(catalogName);
        Table table = select.table(catalogName, databaseName, tableName);

        return new StringBuilder().append(scriptService.createCatalog(catalog, false))
                .append("\n")
                .append(scriptService.dropTable(table))
                .append("\n")
                .append(scriptService.createTable(table))
                .append("\n")
                .append(scriptService.selectCountTable(table))
                .append("\n")
                .append(scriptService.selectTable(table))
                .append("\n")
                .toString();
    }

    public String createCatalog(Catalog catalog, boolean test) {
        String dorisCatalogName = catalog.getDorisCatalogName();
        if (StringUtils.isBlank(dorisCatalogName)) {
            throw new RuntimeException("[Source] " + catalog.getCatalogName() + " -> CREATE CATALOG -> 异常 => dorisCatalogName=" + dorisCatalogName);
        }

        String catalogName = (test ? Constant.TEST_CATALOG_PREFIX : "") + dorisCatalogName;

        return String.format("""
                        CREATE CATALOG IF NOT EXISTS `%s` COMMENT '%s'
                        PROPERTIES (
                            "type"="jdbc",
                            "driver_url" = "%s",
                            "driver_class" = "%s",
                            "jdbc_url" = "%s",
                            "user"="%s",
                            "password"="%s",
                            "lower_case_meta_names" = "false",
                            "meta_names_mapping" = "",
                            "only_specified_database" = "false",
                            "include_database_list" = "",
                            "exclude_database_list" = "",
                            "connection_pool_min_size" = "%d",
                            "connection_pool_max_size" = "%d",
                            "connection_pool_max_wait_time" = "%d",
                            "connection_pool_max_life_time" = "%d",
                            "connection_pool_keep_alive" = "%b",
                            "enable.auto.analyze" ="false"
                        );"""
                , catalogName
                , catalog.getCatalogName()
                , dorisProperties.getDriverJar(catalog.getDatabaseType())
                , dorisProperties.getDriverClass(catalog.getDatabaseType())
                , catalog.getUrl()
                , catalog.getUserName()
                , CryptoUtils.decrypt(catalog.getPassword())
                , dorisProperties.getCatalog().getConnectionPool().getMinSize()
                , dorisProperties.getCatalog().getConnectionPool().getMaxSize()
                , dorisProperties.getCatalog().getConnectionPool().getMaxWaitTime()
                , dorisProperties.getCatalog().getConnectionPool().getMaxLifeTime()
                , dorisProperties.getCatalog().getConnectionPool().getKeepAlive()
        );
    }

    public String selectTable(Table table) {
        return "SELECT * FROM %s.%s;".formatted(ConfigPropertyHelper.ods(), table.getTransTableName());
    }

    public String selectCountTable(Table table) {
        return "SELECT count(*) FROM %s.%s; -- %s".formatted(ConfigPropertyHelper.ods(), table.getTransTableName(), table.getRowCount());
    }

    public String dropTable(Table table) {
        return "DROP TABLE IF EXISTS %s.%s;".formatted(ConfigPropertyHelper.ods(), table.getTransTableName());
    }

    public String createTable(Table table) {
        String template = """
                CREATE TABLE IF NOT EXISTS %s.%s(
                %s
                ) UNIQUE KEY(%s)
                DISTRIBUTED BY HASH(%s) BUCKETS %s
                COMMENT '%s'
                PROPERTIES (
                    "replication_num" = "3",
                    "is_being_synced" = "false",
                    "compression" = "LZ4",
                    "enable_unique_key_merge_on_write" = "true",
                    "light_schema_change" = "true",
                    "skip_write_index_on_load" = "true",
                    "enable_mow_light_delete" = "false",
                    "enable_unique_key_skip_bitmap_column" = "true",
                    "store_row_column" = "true",
                    "disable_storage_row_cache" = "true"
                );""";


        String ods = ConfigPropertyHelper.ods();
        String tableName = table.getTransTableName();
        String columnDefinition = columnDefinitions(table);
        String uniqueKeys = uniqueKeys(table);
        Integer bucket = table.getTransBucketCount();
        String comment = table.getTransComment();
        return template.formatted(ods, tableName, columnDefinition, uniqueKeys, uniqueKeys, bucket, comment);
    }

    public String uniqueKeys(Table table) {
        return select.columnsWithDimension(table).stream().map(Column::getTransColumnName).collect(Collectors.joining(","));
    }

    public String columnDefinitions(Table table) {
        // 维度列（需要优先级）
        List<Column> dimensions = select.columnsWithDimension(table);
        // 非维度列
        List<Column> columns = select.columnsWithoutDimension(table);

        // 列总量
        int columnCount = dimensions.size() + columns.size();

        // 构造列表
        List<String> columnNames = new ArrayList<>(columnCount);
        List<String> columnTypes = new ArrayList<>(columnCount);
        List<String> columnComments = new ArrayList<>(columnCount);
        for (Column column : dimensions) {
            columnNames.add(column.getTransColumnName());
            columnTypes.add(column.transFullDataType());
            columnComments.add(column.getTransComment());
        }
        for (Column column : columns) {
            columnNames.add(column.getTransColumnName());
            columnTypes.add(column.transFullDataType());
            columnComments.add(column.getTransComment());
        }

        // 构造 gather_time
        columnNames.add("gather_time");
        columnTypes.add(DorisDataType.DATETIME.toString());
        columnComments.add("采集时间");

        // 计算最大长度（用于格式化）
        int maxColumnName = columnNames.stream().mapToInt(String::length).max().orElse(0);
        int maxColumnType = columnTypes.stream().mapToInt(String::length).max().orElse(0);

        StringBuilder columnDefinition = new StringBuilder();
        for (int i = 0; i < columnCount; i++) {
            String columnName = columnNames.get(i);
            String columnType = columnNames.get(i);
            String columnComment = columnNames.get(i);
            if (i < columnCount - 1) {
                columnDefinition.append(("    %-" + maxColumnName + "s %-" + maxColumnType + "s COMMENT '%s',\n").formatted(columnName, columnType, columnComment));
            } else {
                // gather_time
                columnDefinition.append(("    %-" + maxColumnName + "s %-" + maxColumnType + "s DEFAULT CURRENT_TIMESTAMP COMMENT '%s'").formatted(columnName, columnType, columnComment));
            }
        }
        return columnDefinition.toString();
    }

    public String seatunnelScript(String catalogName, String databaseName, String tableName) {
        Catalog catalog = select.catalog(catalogName);
        Table table = select.table(catalogName, databaseName, tableName);

        StringBuilder script = new StringBuilder();

        // env
        script.append("""
                env {
                  job.mode = "BATCH"
                  shade.identifier = "base64"
                }
                
                """);

        // Source
        script.append("""
                source {
                  Jdbc {
                    result_table_name = "source_data"
                    url = "%s"
                    driver = %s
                    user = "%s"
                    password = "%s"
                    connection_check_timeout_sec = 100000
                    fetch_size = 5000
                    query = "SELECT * FROM %s.%s"
                    #where_condition = "WHERE 1 = 1"
                  }
                }
                
                """.formatted(catalog.getUrl(),
                catalog.getDriverClass(),
                catalog.getUserName(),
                catalog.getPassword(),
                table.getDatabaseName(),
                table.getTableName()));

        // Transform
        script.append("""
                transform {
                  FieldRename {
                    plugin_input = "source_data"
                    convert_case = "LOWER"
                    plugin_output = "lower_field_transform_data"
                  }
                  SQL {
                    source_table_name = "lower_field_transform_data"
                    result_table_name = "out_data"
                    query = "SELECT * FROM lower_field_transform_data"
                    plugin_output = "transform_data"
                  }
                }
                
                """);

        // Sink
        script.append("""
                sink {
                   Doris {
                    source_table_name = "transform_data"
                    fenodes = "%s:%s"
                    username = "%s"
                    password = "%s"
                    database = %s
                    doris.batch.size = 20000
                    data_save_mode = "APPEND_DATA"
                    table = "%s"
                    sink.label-prefix = "%s"
                    sink.enable-2pc = "false"
                    doris.config {
                      format = "json"
                      read_json_by_line = "true"
                    }
                  }
                }
                """.formatted("",
                "",
                "",
                "",
                ConfigPropertyHelper.ods(),
                table.getTransTableName(),
                table.getTransTableName()));

        return script.toString();
    }

    public void generateDolphinSchedulerTask(String catalogName, String databaseName) {

    }
}